package com.源码分析;

import java.util.Arrays;

import reactor.core.publisher.Flux;

public class IndexDemo {

	public static void main(String[] args) {
		// index生成索引号+真实数据的tuple2
		Flux.range(1, 5).index().subscribe(x -> {
			System.out.println(x.getT1() + ", " + x.getT2());
		});

		int bucketSize = 5;
		Flux.range(1, 6)
			.index()
			.scan(new int[bucketSize], (acc, elem) -> {acc[(int)(elem.getT1() % bucketSize)] = elem.getT2(); return acc;})
			.skip(bucketSize) // 跳过流开头的一些元素来收集计算移动平均所需的足够数据
			.map(array -> Arrays.stream(array).sum() * 1.0 / bucketSize)
			.subscribe(System.out::println);
	}

}
